package realtime.dws;

import org.apache.flink.table.api.Table
import BaseConfig._

/**
 * DWS层：数据仓库汇总层，负责特征计算与多源数据融合
 * 工单编号：大数据-用户画像-11-达摩盘基础特征
 */
object DwsLayer {
        /**
         * 汇总年龄标签各维度得分：类目偏好、品牌偏好、价格敏感度等
         * 权重配置：类目30%、品牌20%、价格15%、时间10%、搜索10%、社交10%、设备5%{insert\_element\_15\_}
         */
        def aggregateAgeFeatures(tEnv: StreamTableEnvironment): Table = {
        tEnv.sqlQuery(
        s"""
         |-- 1. 类目偏好得分（按年龄段权重汇总）
         |WITH category_preference AS (
         |  SELECT
         |    ub.user_id,
         |    SUM(CASE WHEN dc.category_name = '潮流服饰' THEN ub.behavior_cnt * 0.9 ELSE 0 END) AS cat18_24,
         |    SUM(CASE WHEN dc.category_name = '潮流服饰' THEN ub.behavior_cnt * 0.8 ELSE 0 END) AS cat25_29,
         |    SUM(CASE WHEN dc.category_name = '潮流服饰' THEN ub.behavior_cnt * 0.6 ELSE 0 END) AS cat30_34,
         |    SUM(CASE WHEN dc.category_name = '潮流服饰' THEN ub.behavior_cnt * 0.4 ELSE 0 END) AS cat35_39,
         |    SUM(CASE WHEN dc.category_name = '潮流服饰' THEN ub.behavior_cnt * 0.2 ELSE 0 END) AS cat40_49,
         |    SUM(CASE WHEN dc.category_name = '潮流服饰' THEN ub.behavior_cnt * 0.1 ELSE 0 END) AS cat50_plus,
         |    SUM(CASE WHEN dc.category_name = '家居用品' THEN ub.behavior_cnt * 0.2 ELSE 0 END) AS cat18_24_2,
         |    -- ... 省略其他类目（家居用品/健康食品）的各年龄段得分计算（参考{insert\_element\_16\_}）
         |    SUM(ub.behavior_cnt) AS total_category_cnt -- 总类目行为数（用于归一化）
         |  FROM (
         |    -- 统计用户各品类行为频次
         |    SELECT user_id, category_id, COUNT(*) AS behavior_cnt
         |    FROM dwd_clean_user_behavior
         |    GROUP BY user_id, category_id
         |  ) ub
         |  JOIN dim_category dc ON ub.category_id = dc.category_id
         |  GROUP BY ub.user_id
         |),
         |-- 2. 品牌偏好得分（参考{insert\_element\_17\_}）
         |brand_preference AS (
         |  SELECT
         |    ub.user_id,
         |    SUM(CASE WHEN db.brand_name = 'ZARA' THEN ub.behavior_cnt * 0.9 ELSE 0 END) AS brand18_24,
         |    SUM(CASE WHEN db.brand_name = 'ZARA' THEN ub.behavior_cnt * 0.7 ELSE 0 END) AS brand25_29,
         |    -- ... 省略海澜之家等品牌的各年龄段得分计算
         |    SUM(ub.behavior_cnt) AS total_brand_cnt
         |  FROM (
         |    SELECT user_id, brand_id, COUNT(*) AS behavior_cnt
         |    FROM dwd_clean_user_behavior
         |    GROUP BY user_id, brand_id
         |  ) ub
         |  JOIN dim_brand db ON ub.brand_id = db.brand_id
         |  GROUP BY ub.user_id
         |),
         |-- 3. 其他维度得分（价格敏感度/时间行为/搜索词/社交/设备，逻辑类似上述）
         |price_sensitivity AS (/* 参考{insert\_element\_18\_}计算各年龄段得分 */),
         |time_behavior AS (/* 参考{insert\_element\_19\_}计算各年龄段得分 */),
         |search_analysis AS (/* 参考{insert\_element\_20\_}计算各年龄段得分 */),
         |social_interaction AS (/* 参考{insert\_element\_21\_}计算各年龄段得分 */),
         |device_info AS (/* 参考{insert\_element\_22\_}计算各年龄段得分 */)
         |-- 汇总各维度得分（按权重加权）
         |SELECT
         |  cp.user_id,
         |  -- 18-24岁总分 = 类目*30% + 品牌*20% + 价格*15% + ... + 设备*5%
         |  ROUND(
         |    (cp.cat18_24 / cp.total_category_cnt) * 0.3 +
         |    (bp.brand18_24 / bp.total_brand_cnt) * 0.2 +
         |    ps.price18_24 * 0.15 +
         |    tb.time18_24 * 0.1 +
         |    sa.search18_24 * 0.1 +
         |    si.social18_24 * 0.1 +
         |    di.device18_24 * 0.05,
         |    4
         |  ) AS age_score_18_24,
         |  -- 省略25-29岁、30-34岁...50岁以上的总分计算（参考{insert\_element\_23\_}）
         |  ROUND(
         |    (cp.cat50_plus / cp.total_category_cnt) * 0.3 +
         |    (bp.brand50_plus / bp.total_brand_cnt) * 0.2 +
         |    ps.price50_plus * 0.15 +
         |    tb.time50_plus * 0.1 +
         |    sa.search50_plus * 0.1 +
         |    si.social50_plus * 0.1 +
         |    di.device50_plus * 0.05,
         |    4
         |  ) AS age_score_50_plus
         |FROM category_preference cp
         |JOIN brand_preference bp ON cp.user_id = bp.user_id
         |JOIN price_sensitivity ps ON cp.user_id = ps.user_id
         |JOIN time_behavior tb ON cp.user_id = tb.user_id
         |JOIN search_analysis sa ON cp.user_id = sa.user_id
         |JOIN social_interaction si ON cp.user_id = si.user_id
         |JOIN device_info di ON cp.user_id = di.user_id
       """.stripMargin)
        }

        /**
         * 汇总性别标签得分：女性/男性品类行为权重计算
         * 行为权重：购买50%、加购/收藏30%、浏览20%{insert\_element\_24\_}
         */
        def aggregateGenderFeatures(tEnv: StreamTableEnvironment): Table = {
        tEnv.sqlQuery(
        s"""
         |-- 定义男女相关品类（参考{insert\_element\_25\_}）
         |WITH gender_category AS (
         |  SELECT category_id, 'female' AS gender_type FROM dim_female_category
         |  UNION ALL
         |  SELECT category_id, 'male' AS gender_type FROM dim_male_category
         |  UNION ALL
         |  SELECT category_id, 'family' AS gender_type FROM dim_family_category
         |),
         |-- 计算用户各品类行为权重
         |user_behavior_weight AS (
         |  SELECT
         |    ub.user_id,
         |    gc.gender_type,
         |    SUM(
         |      CASE ub.behavior_type
         |        WHEN 'buy' THEN 1 * 0.5 -- 购买权重50%
         |        WHEN 'cart' OR ub.behavior_type = 'collect' THEN 1 * 0.3 -- 加购/收藏权重30%
         |        WHEN 'view' THEN 1 * 0.2 -- 浏览权重20%
         |        ELSE 0
         |      END
         |    ) AS behavior_weight
         |  FROM dwd_clean_user_behavior ub
         |  JOIN gender_category gc ON ub.category_id = gc.category_id
         |  GROUP BY ub.user_id, gc.gender_type
         |)
         |-- 汇总女性/男性/家庭品类总权重
         |SELECT
         |  user_id,
         |  COALESCE(SUM(CASE WHEN gender_type = 'female' THEN behavior_weight ELSE 0 END), 0) AS female_score,
         |  COALESCE(SUM(CASE WHEN gender_type = 'male' THEN behavior_weight ELSE 0 END), 0) AS male_score,
         |  COALESCE(SUM(CASE WHEN gender_type = 'family' THEN behavior_weight ELSE 0 END), 0) AS family_score,
         |  COALESCE(SUM(behavior_weight), 0) AS total_gender_score -- 总权重（用于占比计算）
         |FROM user_behavior_weight
         |GROUP BY user_id
       """.stripMargin)
        }

        /**
         * 融合身高数据：多源加权平均，标准差>5cm时取众数
         * 融合规则：按数据源权重加权，同权重取最新数据{insert\_element\_26\_}
         */
        def fuseHeightData(tEnv: StreamTableEnvironment): Table = {
        tEnv.sqlQuery(
        s"""
         |WITH height_stats AS (
         |  SELECT
         |    user_id,
         |    height_cm,
         |    COUNT(*) OVER (PARTITION BY user_id, height_cm) AS height_cnt, -- 计算频次（用于众数）
         |    STDDEV(height_cm) OVER (PARTITION BY user_id) AS height_std, -- 计算标准差
         |    SUM(height_cm * source_weight) OVER (PARTITION BY user_id) AS weighted_sum,
         |    SUM(source_weight) OVER (PARTITION BY user_id) AS weight_total,
         |    ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY data_time DESC) AS rn
         |  FROM dwd_clean_height
         |)
         |-- 融合逻辑：标准差>5cm取众数，否则取加权平均
         |SELECT
         |  user_id,
         |  CASE
         |    WHEN height_std > 5 THEN
         |      (SELECT height_cm FROM height_stats hs2
         |       WHERE hs2.user_id = hs1.user_id
         |       ORDER BY height_cnt DESC LIMIT 1)
         |    ELSE ROUND(weighted_sum / weight_total, 0) -- 加权平均取整（cm）
         |  END AS fused_height_cm,
         |  CURRENT_DATE() AS update_date
         |FROM height_stats hs1
         |WHERE rn = 1 -- 同权重取最新数据
       """.stripMargin)
        }

        /**
         * 融合体重数据：取权重最高数据源，四舍五入保留1位小数
         * 融合规则：设备（1.0）>订单（0.8）>主动填写（0.6）>活动（0.4）{insert\_element\_27\_}
         */
        def fuseWeightData(tEnv: StreamTableEnvironment): Table = {
        tEnv.sqlQuery(
        s"""
         |WITH weight_ranked AS (
         |  SELECT
         |    user_id,
         |    weight_kg,
         |    data_source,
         |    source_weight,
         |    data_time,
         |    -- 按权重降序、时间降序排序，取top1
         |    ROW_NUMBER() OVER (
         |      PARTITION BY user_id
         |      ORDER BY source_weight DESC, data_time DESC
         |    ) AS rn
         |  FROM dwd_clean_weight
         |)
         |-- 输出融合后体重（保留1位小数）
         |SELECT
         |  user_id,
         |  ROUND(weight_kg, 1) AS fused_weight_kg, -- 四舍五入保留1位小数{insert\_element\_28\_}
         |  data_source AS final_source,
         |  CURRENT_DATE() AS update_date
         |FROM weight_ranked
         |WHERE rn = 1
         |UNION ALL
         |-- 无有效数据标记为NULL
         |SELECT DISTINCT
         |  user_id,
         |  NULL AS fused_weight_kg,
         |  'no_valid_data' AS final_source,
         |  CURRENT_DATE() AS update_date
         |FROM dwd_clean_user_behavior
         |WHERE user_id NOT IN (SELECT user_id FROM weight_ranked)
       """.stripMargin)
        }

        // 注册DWS层汇总表
        def registerDwsTables(tEnv: StreamTableEnvironment): Unit = {
        tEnv.createTemporaryView("dws_age_features", aggregateAgeFeatures(tEnv))
        tEnv.createTemporaryView("dws_gender_features", aggregateGenderFeatures(tEnv))
        tEnv.createTemporaryView("dws_fused_height", fuseHeightData(tEnv))
        tEnv.createTemporaryView("dws_fused_weight", fuseWeightData(tEnv))
        }
        }